var Datanode = {
	start: function (host) {
		this.host = host;
		this.socket = this.host.socket(this, 
				this.pipeline, 
				HadoopConf.getValue('dfs.datanode.address'));
		this.host.registerService(this, HadoopConf.getValue("dfs.datanode.ipc.address"));
		
		this.setTimer(0).done(this.begin);
		
	},
	
	begin: function () {
		this.nnAddress = HadoopConf.getValue('dfs.namenode.address');

		this.socket.connect(
			this.nnAddress, 
			HadoopConf.getValue('dfs.namenode.port'));
		
		this.nnipc = HadoopConf.getValue('dfs.namenode.ipc.port');
		
		// register
		this.storageId = this.host.rpc(this.nnAddress, this.nnipc, Namenode.register, this);
		// cannot register --- this datanode is dead
		if (!this.storageId) return;
		
	},
	
	
	pipeline: function (src, sport, dport, packet, size) {
		/* If this is the last closing message */
		if (packet.close) {
			// send ack to namenode now
			this.host.rpc(
				this.nnAddress,
				this.nnipc,
				Namenode.blockRecv,
				[this, packet.block]);
		} else {

			
			/* Write to disk.. send ack when done */
			// If the file does not exists, create one.
			if  (!this.host.disk.fileExists(packet.block.id)) {
				this.host.disk.create(packet.block.id);
			}
			// See if we can write to file
			// if not (e.g. disk full, failed etc).. we don't send an ack
			var ro = this.host.disk.append(packet.block.id, packet.size);
			if (ro) {
				ro.done(function (packet) {
					// send the ack to client
					this.socket.sendto(packet.src, packet.sport, packet, 16);
					// if this block is completed, tell namenode too
					if (packet.block.file.blocksize 
						=== this.host.disk.getFilesize(packet.block.id)) {
						this.host.rpc(
							this.nnAddress,
							this.nnipc,
							Namenode.blockRecv,
							[this, packet.block]);
					}
				}, this, packet);
			}

		}
		
		/* Pipeline to other datanodes */
		// see if this packet must be sent to other datanodes
		var dn = packet.block.datanodes;
		var idx = dn.indexOf(this.host.ip);
		if (idx < dn.length - 1) {
			this.socket.sendto(dn[idx + 1], dport, packet, packet.size);
		}
	},
	
	/** RPC methods */
	getBlock: function (src, port, block, offset, length) {
		this.log(this.host.ip + " asked to send block " + block.id 
		+ " to " + src + ":" + port
		+ " from " + offset + " length " + length);
		
		var remaining = length;
		while (remaining != 0) {
			var to_read = remaining > PACKET_SIZE ? PACKET_SIZE : remaining;
			this.host.disk.read(block.id, to_read).done(function (src, port, length) {
				this.socket.sendto(src, port, null, length);
			}, this, [src, port, to_read]);
			remaining -= to_read;
		}
	}
};